/*
 *    Copyright 2014-2017 Alexey Danilov
 *    Copyright 2021 Institute of Software Chinese Academy of Sciences, ISRC

 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package com.danikula.videocache;

import com.danikula.videocache.interfacers.Cache;
import com.danikula.videocache.interfacers.CacheListener;
import com.danikula.videocache.interfacers.Source;
import com.danikula.videocache.util.LogUtil;
import com.danikula.videocache.util.ProxyCacheUtils;

import ohos.agp.utils.TextTool;

import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Locale;
import java.util.concurrent.atomic.AtomicInteger;

import static com.danikula.videocache.Preconditions.checkNotNull;
import static com.danikula.videocache.util.ProxyCacheUtils.DEFAULT_BUFFER_SIZE;

/**
 * Proxy for {@link Source} with caching support ({@link Cache}).
 * <p/>
 * Can be used only for sources with persistent data (that doesn't change with time).
 * Method {@link #read(byte[], long, int)} will be blocked while fetching data from source.
 * Useful for streaming something with caching e.g. streaming video/audio etc.
 *
 * @author Alexey Danilov ().
 */
public abstract class ProxyCache {
    /**
     * 日志打印工具
     */
    private static final String TAG = "ProxyCache";
    /**
     * 网络请求错误重连次数
     */
    protected static final int MAX_READ_SOURCE_ATTEMPTS = 1;

    /**
     * http请求接口及具体实现类
     */
    protected final Source source;
    /**
     * 缓存读写接口及具体实现类
     */
    protected final Cache cache;
    /**
     * 读写锁
     */
    protected final Object wc = new Object();
    /**
     * 读写停止锁
     */
    protected final Object stopLock = new Object();
    /**
     * 网络请求错误数
     */
    protected final AtomicInteger readSourceErrorsCount;
    /**
     * 网络数据请求及写入缓存线程
     */
    protected volatile Thread sourceReaderThread;
    /**
     * 网络请求及缓存数据写入是否停止
     */
    protected volatile boolean stopped;
    /**
     * 缓存进度
     */
    protected volatile int percentsAvailable = -1;
    /**
     * 缓存进度回调监听器
     */
    protected CacheListener listener;

    public ProxyCache(Source source, Cache cache) {
        this.source = checkNotNull(source);
        this.cache = checkNotNull(cache);
        this.readSourceErrorsCount = new AtomicInteger();
    }

    /**
     * 请求代理分发
     *
     * @param request 请求
     * @param socket  客户端socket
     * @throws IOException          IO错误
     * @throws ProxyCacheException  代理缓存错误
     * @throws InterruptedException 线程中断错误
     */
    public void processRequest(GetRequest request, Socket socket) throws IOException, ProxyCacheException,
            InterruptedException {
    }

    /**
     * 注册缓存进度监听器
     *
     * @param cacheListener 缓存进度监听器
     */
    public void registerCacheListener(CacheListener cacheListener) {
        this.listener = cacheListener;
    }

    /**
     * 读取缓存
     *
     * @param buffer 缓冲区
     * @param offset 读取起始点
     * @param length 缓冲区长度
     * @return 缓存数据
     * @throws ProxyCacheException 错误信息
     */
    public int read(byte[] buffer, long offset, int length) throws ProxyCacheException {
        return -1;
    }

    /**
     * 边请求数据边缓存 并读取缓存文件数据返回给socket
     *
     * @param out    输出流
     * @param offset 视频数据读取起点
     * @throws ProxyCacheException 代理缓存错误
     * @throws IOException         IO错误
     */
    protected void responseWithCache(OutputStream out, long offset) throws ProxyCacheException, IOException {
        byte[] buffer = new byte[DEFAULT_BUFFER_SIZE];
        long tempOffset = offset;
        int readBytes;
        while ((readBytes = read(buffer, tempOffset, buffer.length)) != -1) {
            out.write(buffer, 0, readBytes);
            tempOffset += readBytes;
        }
        out.flush();
    }

    /**
     * 碎片文件下载合并完毕之后 重命名到了原库的缓存位置
     * 这个时候视频数据来源都切换为本地缓存文件了
     *
     * @param out    代理客户端socket输出流
     * @param offset 读取起点
     * @throws ProxyCacheException 错误信息
     * @throws IOException         错误信息
     */
    protected void responseWithFile(OutputStream out, long offset) throws ProxyCacheException, IOException {
        byte[] buffer = new byte[DEFAULT_BUFFER_SIZE];
        int readBytes;
        long tempOffset = offset;
        while ((readBytes = readCacheFile(buffer, tempOffset, buffer.length)) != -1) {
            out.write(buffer, 0, readBytes);
            tempOffset += readBytes;
        }
        out.flush();
    }

    /***
     * 只读缓存文件
     * @param buffer 读取byte数组
     * @param offset 读取起点
     * @param length 读取长度
     * @return 读取的数据
     * @throws ProxyCacheException 错误信息
     */
    protected int readCacheFile(byte[] buffer, long offset, int length) throws ProxyCacheException {
        ProxyCacheUtils.assertBuffer(buffer, offset, length);
        return cache.read(buffer, offset, length);
    }

    /**
     * 检查写入线程错误数
     *
     * @throws ProxyCacheException 代理缓存错误
     */
    protected void checkReadSourceErrorsCount() throws ProxyCacheException {
        int errorsCount = readSourceErrorsCount.get();
        if (errorsCount >= MAX_READ_SOURCE_ATTEMPTS) {
            readSourceErrorsCount.set(0);
            throw new ProxyCacheException("Error reading source " + errorsCount + " times");
        }
    }

    /**
     * 本地拼接resoponse头部信息
     *
     * @param request Model for Http GET request.
     * @return resoponse头部信息
     * @throws ProxyCacheException 代理缓存错误
     * @throws IOException         IO错误
     */
    protected String newResponseHeaders(GetRequest request) throws IOException, ProxyCacheException {
        String mime = source instanceof HttpUrlSource ? ((HttpUrlSource) source).getMime() : "";
        boolean mimeKnown = !TextTool.isNullOrEmpty(mime);
        long length = cache.isCompleted() ? cache.available() : source.length();
        boolean lengthKnown = length >= 0;
        long contentLength = request.partial ? length - request.rangeOffset : length;
        boolean addRange = lengthKnown && request.partial;
        return new StringBuilder()
                .append(request.partial ? "HTTP/1.1 206 PARTIAL CONTENT" + System.lineSeparator() :
                        "HTTP/1.1 200 OK" + System.lineSeparator())
                .append("Accept-Ranges: bytes")
                .append(System.lineSeparator())
                .append(lengthKnown ? format("Content-Length: %d" + System.lineSeparator(), contentLength) : "")
                .append(addRange ?
                        format("Content-Range: bytes %d-%d/%d" + System.lineSeparator(), request.rangeOffset,
                                length - 1, length) : "")
                .append(mimeKnown ? format("Content-Type: %s" + System.lineSeparator(), mime) : "")
                .append(System.lineSeparator()) // headers end
                .toString();
    }

    /**
     * 格式化字符串
     *
     * @param pattern 匹配模式
     * @param args    参数
     * @param <E>     参数类型
     * @return 格式化后的字符串
     */
    protected <E> String format(String pattern, E args) {
        return String.format(Locale.US, pattern, args);
    }

    /**
     * 格式化字符串
     *
     * @param pattern 匹配模式
     * @param args    参数
     * @param args1   参数
     * @param args2   参数
     * @param <E>     参数类型
     * @return 格式化后的字符串
     */
    protected <E> String format(String pattern, E args, E args1, E args2) {
        return String.format(Locale.US, pattern, args, args1, args2);
    }

    /**
     * 中断写入线程
     * 关闭文件流
     */
    public void shutdown() {
        synchronized (stopLock) {
            LogUtil.debug(TAG, "Shutdown proxy for " + source.toString());
            try {
                stopped = true;
                if (sourceReaderThread != null) {
                    sourceReaderThread.interrupt();
                }
                cache.close();
            } catch (ProxyCacheException e) {
                onError(e);
            }
        }
    }

    /**
     * 缓存文件可读取数据不足 读取线程等待写入线程请求网络数据并写入
     *
     * @throws ProxyCacheException 代理缓存错误
     */
    protected void waitForSourceData() throws ProxyCacheException {
        synchronized (wc) {
            try {
                wc.wait(1000);
            } catch (InterruptedException e) {
                throw new ProxyCacheException("Waiting source data is interrupted!", e);
            }
        }
    }

    /**
     * 开启写入线程
     *
     * @throws ProxyCacheException 代理缓存错误
     */
    protected synchronized void readSourceAsync() throws ProxyCacheException {
        boolean readingInProgress =
                sourceReaderThread != null && sourceReaderThread.getState() != Thread.State.TERMINATED;
        if (!stopped && !cache.isCompleted() && !readingInProgress) {
            sourceReaderThread = new Thread(getRunable(), "Source reader for " + source);
            sourceReaderThread.start();
        }
    }

    /**
     * 获取写入线程 Runnable
     *
     * @return 写入线程 Runnable
     */
    protected abstract Runnable getRunable();

    /**
     * 缓存进度回调 并唤醒读取线程(如果之前缓存数据不足 处于等待状态)
     *
     * @param cacheAvailable  缓存文件长度
     * @param sourceAvailable 视频总长度
     */
    protected void notifyNewCacheDataAvailable(long cacheAvailable, long sourceAvailable) {
        onCacheAvailable(cacheAvailable, sourceAvailable);
        synchronized (wc) {
            wc.notifyAll();
        }
    }

    /**
     * 默认的下载进度回调
     *
     * @param cacheAvailable 缓存文件长度
     * @param sourceLength   视频总长度
     */
    protected void onCacheAvailable(long cacheAvailable, long sourceLength) {
        boolean zeroLengthSource = sourceLength == 0;
        int percents = zeroLengthSource ? 100 : (int) ((float) cacheAvailable / sourceLength * 100);
        boolean percentsChanged = percents != percentsAvailable;
        boolean sourceLengthKnown = sourceLength >= 0;
        if (sourceLengthKnown && percentsChanged) {
            onCachePercentsAvailableChanged(percents);
        }
        percentsAvailable = percents;
    }

    /**
     * 分片/分节点缓存 缓存进度回调
     *
     * @param percentsAvailable 下载百分比
     */
    protected void onCachePercentsAvailableChanged(int percentsAvailable) {
    }

    /**
     * 本次下载流程完毕 刷新缓存进度
     */
    public void onSourceRead() {
        // guaranteed notify listeners after source read and cache completed
        percentsAvailable = 100;
        onCachePercentsAvailableChanged(percentsAvailable);
    }

    /**
     * 检查视频文件是否下载完毕 缓存文件是否需要重命名
     *
     * @throws ProxyCacheException 代理缓存错误
     */
    protected void tryComplete() throws ProxyCacheException {
        synchronized (stopLock) {
            if (!isStopped() && cache.available() == source.length()) {
                cache.complete();
            }
        }
    }

    /**
     * 本次请求缓存流程结束后的操作
     *
     * @throws ProxyCacheException 代理错误
     */
    public void doComplete() throws ProxyCacheException {
        // do nothing
    }

    protected boolean isStopped() {
        return Thread.currentThread().isInterrupted() || stopped;
    }

    /**
     * 关闭网络请求资源
     */
    protected void closeSource() {
        try {
            source.close();
        } catch (ProxyCacheException e) {
            onError(new ProxyCacheException("Error closing source " + source, e));
        }
    }

    /**
     * 打印错误信息
     *
     * @param throwable 错误信息
     */
    protected final void onError(final Throwable throwable) {
        boolean interruption = throwable instanceof InterruptedProxyCacheException;
        if (interruption) {
            LogUtil.debug(TAG, "ProxyCache is interrupted");
        } else {
            LogUtil.error(TAG, "ProxyCache error" + throwable.toString());
        }
    }
}
